package com.talent.service.iot.mqtt.handler;

import com.talent.service.iot.mqtt.config.Constant;
import com.talent.service.iot.mqtt.model.fix.TlMqttFixedHead;
import com.talent.service.iot.mqtt.model.payload.TlMqttPublishPayload;
import com.talent.service.iot.mqtt.model.request.TlMqttPubRecReq;
import com.talent.service.iot.mqtt.model.request.TlMqttPublishReq;
import com.talent.service.iot.mqtt.model.response.TlMqttPubAck;
import com.talent.service.iot.mqtt.model.variable.TlMqttPublishVariableHead;
import com.talent.service.iot.mqtt.service.IMqttMsgService;
import com.talent.service.iot.mqtt.service.IMqttRetainMsgService;
import com.talent.service.iot.mqtt.service.IMqttSessionService;
import com.talent.service.iot.mqtt.service.IMqttSubscriptionService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.util.AttributeKey;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

import java.util.HashSet;
import java.util.Set;

/**
 * @ClassName：ConnectHandler
 * @Author: hszhou
 * @Date: 2024/11/25 11:06
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@ChannelHandler.Sharable
@Component
public class TlMqttPublishHandler extends SimpleChannelInboundHandler<TlMqttPublishReq> {

    private static final  Logger logger = LoggerFactory.getLogger(TlMqttPublishHandler.class);

    private final IMqttSubscriptionService subscriptionService;

    private final IMqttRetainMsgService retainMsgService;

    private final IMqttMsgService msgService;

    private final IMqttSessionService sessionService;

    /**用于borker存储收到的qos2的消息id*/
    public static Set<Integer> qos2PacketIds = new HashSet<>();


    public TlMqttPublishHandler(IMqttSubscriptionService subscriptionService,
                                IMqttRetainMsgService retainMsgService,
                                IMqttMsgService msgService,
                                IMqttSessionService sessionService){
        this.subscriptionService = subscriptionService;
        this.retainMsgService = retainMsgService;
        this.msgService = msgService;
        this.sessionService=sessionService;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TlMqttPublishReq req) throws Exception {

        logger.debug("in【PUBLISH】 handler");
        Channel channel = ctx.channel();
        Object clientId = channel.attr(AttributeKey.valueOf(Constant.CLIENT_ID)).get();
        if(clientId==null){
            channel.close();
            return;
        }

        TlMqttFixedHead fixedHead = req.getFixedHead();
        TlMqttPublishVariableHead variableHead = req.getVariableHead();
        TlMqttPublishPayload payload = req.getPayload();

        boolean retain = fixedHead.getRetain();
        MqttQoS messageQos = fixedHead.getQos();
        String topic = variableHead.getTopic();
        Integer messageId = variableHead.getMessageId();
        String content = payload.getContent().toString();

        logger.debug("【{}】 send topic 【{}】 qos【{}】 messageId 【{}】",clientId,topic,messageQos,messageId);
        /*如果是保留消息 存储*/

        if (retain) {
            storeRetain(topic, content,messageQos,messageId);
        }
        switch (messageQos){
            //如果收到了qos0的消息 只需要负责转发就可以了
            case AT_MOST_ONCE -> publish(topic,messageQos,content,retain);
            //如果收到了qos1的消息  需要发送ack消息给客户端 然后转发消息
            case AT_LEAST_ONCE -> {
                publish(topic,messageQos,content,retain);
                sendAck(messageId, ctx);
            }
            case EXACTLY_ONCE -> {
                /*
                 * 如果收到了qos2的消息已经存在 那么就直接丢弃
                 * 否则保存起来
                 */
                if (qos2PacketIds.contains(messageId)) {
                    logger.debug("【broker】 0. receive qos2 messageId is dup");
                    return;
                }
                qos2PacketIds.add(messageId);
                logger.debug("【BROKER】 1. receive pub【qos2】 messageId is 【{}】",messageId);

                publish(topic,messageQos,content,retain);
                //发送rec消息给发送者  如果在一定时间内没收到rel消息 那么就需要重发
                sendRec(clientId.toString(), messageId, ctx, req);

            }
            default -> logger.error("error qos level");
        }


    }

    /**
     * 向订阅topic的客户端转发消息
     * @param topic 主题
     * @param pubQos 发布的消息等级
     * @param content 发布的内容
     * @param retain 是否是保留消息
     * @return Void
     */
    public Mono<Void> publish(String topic,MqttQoS pubQos,String content,boolean retain){
        subscriptionService.find(topic)
                .parallel()
                .subscribe(sub->{
                    //订阅者收到mqtt的消息qos级别 最终取决于发布消息的qos和主题的订阅的qos
                    int subQos = sub.getQos();
                    int qos= Math.min(subQos, pubQos.value());

                    MqttQoS mqttQoS = MqttQoS.valueOf(qos);
                    String clientId = sub.getClientId();
                    Integer customMessageId = msgService.getMessageId();
                    TlMqttPublishReq publishReq = TlMqttPublishReq.of(topic,mqttQoS,retain,content,customMessageId);

                    //如果发送的消息等级是1或者2的话 就需要将消息保存起来 方便下次再发送
                    if(mqttQoS==MqttQoS.EXACTLY_ONCE || mqttQoS==MqttQoS.AT_LEAST_ONCE){
                        publishReq.getFixedHead().setDup(true);
                        msgService.savePublishMessage(clientId, customMessageId, publishReq);
                        logger.debug("【CLIENT】 1.【{}】 store messageId 【{}】",clientId,customMessageId);
                    }

                    //找到客户端绑定的session
                    sessionService.find(clientId)
                            .doOnNext(session -> {
                                /*
                                 * 当CleanSession为0的会话断开后，
                                 * 服务器还必须将所有和客户端订阅相关的QoS1和QoS2的消息作为会话状态的一部分存储起来[MQTT-3.1.2-5]。
                                 * 也可以选择把QoS0的消息也存储起来。
                                 */
                                if(session.getCleanSession()==0){
                                    if(mqttQoS==MqttQoS.EXACTLY_ONCE || mqttQoS==MqttQoS.AT_LEAST_ONCE){
                                        Set<Integer> publishMsgStore = session.getPublishMsgStore();
                                        publishMsgStore.add(customMessageId);
                                        logger.debug("client 【{}】 session store messageId is 【{}】",clientId,customMessageId);
                                    }
                                }
                            })
                            .subscribe(session -> {
                                Channel channel = session.getChannel();
                                if(channel!=null && channel.isActive()){
                                    channel.writeAndFlush(publishReq);
                                }
                            });
                });
        return Mono.empty();
    }

    /**
     * 保存保留消息 在新订阅的时候发送
     * @param topic 主题
     * @param content 内容
     */
    private void storeRetain(String topic,String content,MqttQoS qos,Integer messageId) {
        if(StringUtils.isEmpty(content)){
            retainMsgService.clear(topic);
            return;
        }
        TlMqttPublishReq pubMsg = TlMqttPublishReq.of(topic, qos, true, content, messageId);
        retainMsgService.save(topic, pubMsg);
    }


    /**
     * 构建ack消息发送返回
     * @param messageId 消息ID
     * @param ctx 通道
     */
    private void sendAck(int messageId, ChannelHandlerContext ctx){
        TlMqttPubAck res = TlMqttPubAck.of(messageId);
        logger.debug("send ack messageId【{}】",messageId);
        ctx.channel().writeAndFlush(res);
    }

    private void sendRec(String clientId,int messageId,ChannelHandlerContext ctx ,TlMqttPublishReq req){

        msgService.findPublishMessage(clientId,messageId)
                .switchIfEmpty(
                        Mono.fromRunnable(()->{
                            logger.debug("send rec messageId【{}】",messageId);
                            //本地存起来
                            msgService.savePublishMessage(clientId,messageId,req);
                            //发送rec消息给发送者
                            TlMqttPubRecReq res = TlMqttPubRecReq.of(messageId);
                            ctx.channel().writeAndFlush(res);
                        })
                ).subscribe();

    }
}
